package function;

import bean.TableProcess;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class TableProcessFunction extends BroadcastProcessFunction<String, String, JSONObject> {
    Connection connection=null;
    private OutputTag<JSONObject> objectOutputTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(OutputTag<JSONObject> objectOutputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.objectOutputTag = objectOutputTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection("jdbc:phoenix:192.168.1.10:2181");
    }

    @Override
    public void processElement(String value, ReadOnlyContext ctx, Collector<JSONObject> collector) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        String key = jsonObject.getString("tableName")+":"+jsonObject.getString("type");
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(key);
        if(tableProcess!=null){
            jsonObject.put("sinktype",tableProcess.getSinktype());
            jsonObject.put("sinkdatabase",tableProcess.getSinkdbname());
            jsonObject.put("sinktable",tableProcess.getSinktbname());
            if("kafka".equals(jsonObject.getString("sinktype"))){
                collector.collect(jsonObject);
            }else if ("hbase".equals(jsonObject.getString("sinktype"))){
                ctx.output(objectOutputTag,jsonObject);
            }
        }else{
            System.out.println("该配置不存在");
        }
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> collector) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        String after = jsonObject.getString("after");
        TableProcess tableProcess = JSONObject.parseObject(after, TableProcess.class);
        String key = tableProcess.getTbname()+":"+tableProcess.getType();
        PreparedStatement preparedStatement = null;
        System.out.println(tableProcess.getSinktype());
        if("hbase".equals(tableProcess.getSinktype())){
            StringBuffer stringBuffer = new StringBuffer();
            stringBuffer.append("create table if not exists ");
            stringBuffer.append(tableProcess.getSinkdbname()+"."+tableProcess.getSinktbname()+" (");
            String[] split = tableProcess.getColname().split(",");
            for(int i=0;i<split.length;i++){
                if(split[i].equals(tableProcess.getPk())){
                    stringBuffer.append(split[i]+" varchar primary key ");
                }else{
                    stringBuffer.append(split[i]+" varchar ");
                }
                if(i<split.length-1){
                    stringBuffer.append(",");
                }
            }
            stringBuffer.append(")");
            System.out.println("建表语句为："+stringBuffer);
            preparedStatement = connection.prepareStatement(stringBuffer.toString());
            preparedStatement.execute();

        }
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key, tableProcess);
    }
}
